/**
 * @brief 
 * @author zhangpeihao
 */

#pragma once
#include <zsystem/core/networking/socket.h>
#include <zsystem/core/networking/packet.h>
#include <zsystem/core/common/queue.h>
#include <zsystem/core/common/map.h>
#include <zsystem/core/common/lockfree.h>
#include <zsystem/core/common/event_types.h>
#include <zsystem/core/threading/thread.h>
#include <zsystem/core/threading/evthreadpool.h>

#define DEFAULT_MAX_AUTH_MSG_ID 100
#ifdef __DEBUG__
#define MONO_TIMESTAMP_SECOND 10
#define MONO_TIMESTAMP_USECOND 0
#endif
#ifdef __RELEASE__
#define MONO_TIMESTAMP_SECOND 0
#define MONO_TIMESTAMP_USECOND 200000 // 0.2 second
#endif

namespace zsystem {
  enum {
    ERROR_CODE_SUCCESS = 0,
    ERROR_CODE_INTERNAL_ERROR = 1,
    ERROR_CODE_PARAM_ERROR = 2,
    ERROR_CODE_LUA_FAULT = 3,
    ERROR_CODE_WAIT_PROCESS = 100,
    ERROR_CODE_RETRY = 101
  };

  class tcpConnection;
  class listenerContainer;

  typedef void (*msgReceivedListener)(listenerContainer* container, tcpConnection* connection, const Z_U16 seqId, const Z_U16 msgId, char* const buffer, const Z_U16 len);
  class connectionCallbacks{
  protected:
    virtual void onConnected(tcpConnection* connection) = 0;
    virtual void onDisconnected(tcpConnection* connection) = 0;
  };

  class connectionManager{
  public:
    virtual tcpConnection* findConnection(const Z_I32 connectionId) = 0;
    virtual tcpConnection* newConnection() = 0;
  };
#define SEND_BY_CONN_ID(MANA, CONN_ID, MSG_ID, BUFFER, BUFFER_LEN)          \
  do{ zsystem::tcpConnection* conn = (MANA).findConnection(CONN_ID);        \
  conn || conn->send(MSG_ID, BUFFER, BUFFER_LEN); } while(0);
#define RESPONSE_BY_CONN_ID(MANA, CONN_ID, SEQ_ID, MSG_ID, BUFFER, BUFFER_LEN)      \
  do{ zsystem::tcpConnection* conn = (MANA).findConnection(CONN_ID);                \
  conn || conn->response(SEQ_ID, MSG_ID, BUFFER, BUFFER_LEN); } while(0);
#define CLOSE_BY_CONN_ID(MANA, CONN_ID)                                     \
  do{ zsystem::tcpConnection* conn = (MANA).findConnection(CONN_ID);        \
  if ( conn != Z_NULL){ conn->close();} } while(0);

  class listenerContainer {
  public:
    listenerContainer(msgReceivedListener defaultListener, const Z_U16 maxMsgId = 1024)
    : _maxMsgId(maxMsgId)
    , _defaultListener(defaultListener){
      _listenerMap = (msgReceivedListener*)Z_MALLOC(sizeof(msgReceivedListener) * maxMsgId);
      memset(_listenerMap, 0, sizeof(msgReceivedListener) * maxMsgId);
      Z_ASSERT_ALL(_defaultListener != Z_NULL);
    }

    virtual ~listenerContainer() {
      if ( _listenerMap != Z_NULL ) {
        Z_FREE(_listenerMap);
      }
    }

    const Z_BOOL addListener(const Z_U16 msgId, msgReceivedListener listener){
      Z_ASSERT_RETURN( msgId <= _maxMsgId && _listenerMap[msgId] == Z_NULL, Z_FALSE );
      _listenerMap[msgId] = listener;
      return Z_TRUE;
    }
  protected:
    msgReceivedListener getListener(const Z_U16 msgId){
      Z_ASSERT_RETURN( msgId <= _maxMsgId && _listenerMap[msgId] != Z_NULL, Z_NULL );
      return _listenerMap[msgId];
    }
    void doCallback(tcpConnection* connection, const Z_U16 seqId, const Z_U16 msgId, char* const buffer, const Z_U16 len);
    msgReceivedListener* _listenerMap;
    msgReceivedListener _defaultListener;
    Z_U16 _maxMsgId;
  friend class tcpConnection;
  };

  typedef zsystem::stdQueue<zsystem::packet*> sendQueueType;
  class tcpConnection : public connectionCallbacks, noncopyable {
  public:
    const Z_BOOL send(const Z_U16 msgId, const char* buffer, const Z_U16 len);
    const Z_BOOL response(const Z_U16 seqId, const Z_U16 msgId, const char* const buffer, const Z_U16 len);
    virtual void close();
    virtual void cleanup();
    virtual ~tcpConnection();
    void authPass(){
      _authPass = Z_TRUE;
    }
    void authFailed(){
      _authPass = Z_FALSE;
      close();
    }
    const Z_BOOL isAuth() const{
      return _authPass;
    }
    
    Z_SOCKET_ADDRESS& getRemoteAddress() {
      return _socket->getAddress();
    }

    const char* getRemoteAddressString(){
      return _socket->getAddressString();
    }
    const Z_I32 getMark() const {
      return _mark;
    }
    void setMark( const Z_I32 mark ) {
      _mark = mark;
    }
    const Z_U32 getId() const{
      return _id;
    }
  protected:
    tcpConnection(zsystem::listenerContainer* listener, const Z_I32 sendBufferSize, const Z_I32 recieveBufferSize, const Z_I32 keepAliveInterval);
    tcpConnection(zsystem::listenerContainer* listener, const Z_SOCKET_HANDLE& socket, const Z_SOCKET_ADDRESS& address, const Z_I32 sendBufferSize, const Z_I32 recieveBufferSize, const Z_I32 keepAliveInterval);
    tcpConnection(const Z_U32 id, zsystem::listenerContainer* listener, zsystem::tcpSocket* socket);
    static void readCallback(Z_EV_BUFFER_EVENT* ev, void* arg);
    static void writeCallback(Z_EV_BUFFER_EVENT* ev, void* arg);
    static void connectionCallback(Z_EV_BUFFER_EVENT* ev, short what, void* arg);
  private:
    const Z_BOOL read(Z_EV_BUF* input);
    const Z_BOOL write(Z_EV_BUF* output);
  protected:
    zsystem::tcpSocket* _socket;
    zsystem::packet* _receivePacket;
    zsystem::listenerContainer* _listener;
    sendQueueType* _sendQueue;
    Z_EV_BUFFER_EVENT* _evbuf;
    Z_AU16 _seqId;
    Z_I32 _received;
    Z_I32 _mark;
    Z_I32 _removedAt;
    Z_U32 _id;
    Z_BOOL _connected;
    Z_BOOL _authPass;
  };

  class tcpClient : public tcpConnection , public listenerContainer {
  public:
    tcpClient(zsystem::msgReceivedListener defaultListener, const Z_I32 sendBufferSize, const Z_I32 recieveBufferSize,
              Z_EV_BASE* base = Z_NULL);
    virtual ~tcpClient();
    Z_BOOL connect(const char* serverAddress);
    void close();
  protected:
    virtual void onConnected(tcpConnection* connection){}
    virtual void onDisconnected(tcpConnection* connection){}
  private:
    static void threadMain(Z_PTR arg);
  protected:
    zsystem::thread* _thread;
    Z_EV_BASE* _evbase;
    Z_BOOL _newBase;
  };

  class tcpServer;
  class tcpServerConnection;
  typedef zsystem::evThreadM tcpServerThread;
  typedef zsystem::evThreadPoolM tcpServerThreadPool;
  typedef zsystem::stdMap<Z_I32, zsystem::mutexLock> tcpServerConnectionMap;
  typedef zsystem::stdQueue<tcpServerConnection*> tcpServerConnectionQueue;

  class tcpServerConnection : public tcpConnection {
  public:
    virtual ~tcpServerConnection();
    void close();
  protected:
    tcpServerConnection(const Z_U32 id, zsystem::tcpServer* server, zsystem::tcpServerThread* thread, zsystem::tcpSocket* socket);
    void doClose();
  protected:
    virtual void onConnected(tcpConnection* connection){}
    virtual void onDisconnected(tcpConnection* connection){}
    zsystem::tcpServer* _server;
    zsystem::tcpServerThread* _thread;
  friend class zsystem::tcpServer;
  };

  class tcpServer : public listenerContainer, public connectionManager {
  public:
    tcpServer(zsystem::msgReceivedListener defaultListener, const char* bindAddress, const Z_I32 sendBufferSize, const Z_I32 recieveBufferSize, const Z_I32 keepAliveInterval, const Z_U16 threads, const Z_I32 backlog);
    virtual ~tcpServer();
    const Z_BOOL start();
    void stop();
    virtual const Z_BOOL close(const Z_I32 connectionId);
    const Z_I32 connections(){
      return _connectionMap->getNumber();
    }
    zsystem::tcpConnection* findConnection(const Z_I32 connectionId){
      Z_PTR data;
      if ( _connectionMap->lookup(connectionId, data) )
      {
        return static_cast<zsystem::tcpConnection*>(data);
      }
      return Z_NULL;
    }
  private:
    static void threadInit(Z_PTR thread, Z_PTR initarg);
    static void threadAccpetMian(Z_PTR arg);
    static void onNewConnection(Z_PTR thr, Z_PTR arg, Z_PTR initarg);
    static void onAccept(Z_EV_SRV* srv, evutil_socket_t nfd, struct sockaddr *peer_sa, int peer_socklen, Z_PTR arg);
    static void onUpdateTimestamp(Z_SOCKET_HANDLE fd, Z_I16 ev, Z_PTR arg);
  protected:
    zsystem::tcpServerThreadPool* _threadPool;
    char* _bindAddress;
    zsystem::tcpSocket* _listenSocket;
    zsystem::thread* _acceptThread;
    zsystem::tcpServerConnectionMap* _connectionMap;
    zsystem::tcpServerConnectionQueue* _releaseConnectionList;
    Z_EV_BASE* _evbase;
    Z_EV* _timestampEvent;
    Z_I32 _sendBufferSize;
    Z_I32 _recieveBufferSize;
    Z_I32 _keepAliveInterval;
    Z_I32 _backlog;
    Z_I32 _motoTimestamp;
    struct timeval _tv;
    Z_AU32 _connectionIdAlloc;
    Z_U16 _threads;
    Z_AU8 _exit;
  friend class tcpServerConnection;
  };

}
